package com.steven.logserver.mq;

import com.google.common.collect.Maps;
import com.google.protobuf.Descriptors;
import com.rabbitmq.client.Channel;
import com.steven.logserver.protobuf.LogServerSendReqProto;
import com.steven.logserver.utils.ElasticsearchUtil;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Map;

/**
 * 消费者监听器
 *
 * @author qinxiaohu
 */
@Component
public class MsgListener implements ChannelAwareMessageListener {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${log.netty.server.file-folder}")
    private String fileFolder;
    @Value("${elasticsearch.number-of-replicas}")
    private Integer numberOfReplicas;
    @Value("${elasticsearch.number-of-shards}")
    private Integer numberOfShards;

    private DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy_MM_dd");

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {

        byte[] body = message.getBody();
        if(body == null || body.length == 0){
            return;
        }
        Object object = toObject(body);
        if(object == null){
            return;
        }
        LogServerSendReqProto.LogServerSendReq req = (LogServerSendReqProto.LogServerSendReq) object;
        // 1. 日志文件不存在则创建，并使用追加形式的写模式
        String path = fileFolder + "/log_" + LocalDate.now().format(formatter) + ".file";
        try(BufferedWriter writer =
                    Files.newBufferedWriter(Paths.get(path), StandardCharsets.UTF_8, StandardOpenOption.CREATE, StandardOpenOption.APPEND)){
            StringBuilder params = new StringBuilder("{ ");
            Map<Descriptors.FieldDescriptor, Object> allFields = req.getAllFields();
            Map<String, Object> jsonObject = Maps.newHashMap();
            String createTime = "createTime";
            jsonObject.put(createTime, System.currentTimeMillis());
            allFields.forEach((k, v) -> {
                String name = k.getJsonName();
                jsonObject.put(name, v);
                // 判断是protobuf协议格式的枚举，需要转换
                if(v instanceof Descriptors.EnumValueDescriptor){
                    jsonObject.put(name, ((Descriptors.EnumValueDescriptor) v).getName());
                }
                String value = "\"" + v + "\"";
                params.append("\"").append(name).append("\":").append(value).append(", ");
            });
            String substring = params.substring(0, params.length() - 2);
            writer.write(substring + " }" + System.getProperty("line.separator", "\n"));
            // 2. 同步到ES。使用被收集服务的服务名（TIPS：index不能出现大写字母）作为index，判断index是否存在
            // TIPS：6.0的版本不允许一个index下面有多个type，并且官方说是在接下来的7.0版本中会删掉type
            String index = req.getCollectServiceName().toLowerCase();
            if(!ElasticsearchUtil.isIndexExist(index)) {
                //单机直接设置number_of_replicas为0，不需要备份
                Settings.Builder setting = Settings.builder()
                        .put("number_of_shards", numberOfShards).put("number_of_replicas", numberOfReplicas);
                //从LogServerSendReqProto属性中映射ElasticSearch的mapping字段
                XContentBuilder mapping = XContentFactory.jsonBuilder().startObject().startObject("properties");
                mapping.startObject(createTime).field("type", "long").endObject();
                allFields.forEach((k, v) -> {
                    String name = k.getJsonName();
                    try {
                        // 判断是protobuf协议格式的枚举，需要转换
                        if(v instanceof Descriptors.EnumValueDescriptor){
                            mapping.startObject(name).field("type", "keyword").endObject();
                        }
                        else if(v instanceof Integer){
                            mapping.startObject(name).field("type", "integer").endObject();
                        }
                        else if(v instanceof Long){
                            mapping.startObject(name).field("type", "long").endObject();
                        }
                        else if(v instanceof String){
                            mapping.startObject(name).field("type", "keyword").endObject();
                        } else{
                            logger.error("不匹配类型：{}", name);
                        }
                    } catch (IOException e) {
                        logger.error("封装es mapping 异常：{}", e.getMessage());
                    }
                });
                mapping.endObject().endObject();
                ElasticsearchUtil.addMapping(index, index, mapping, setting);
            }
            // 同步到ES
            ElasticsearchUtil.addData(jsonObject, index, index, null);
        } catch (IOException e) {
            logger.error("记录日志失败：{}", e.getMessage());
        }finally {
            // 当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令，然后RabbitMQ才会将消息删除
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    private Object toObject(byte[] bytes){

        Object object = null;
        try (ByteArrayInputStream arrayInputStream = new ByteArrayInputStream(bytes);
             ObjectInputStream inputStream = new ObjectInputStream(arrayInputStream)){
            // 从objectInputStream流中读取一个对象
            object = inputStream.readObject();
        } catch(ClassNotFoundException | IOException e) {
           logger.error("字节数组转对象失败：{}", e.getMessage());
        }

        return object;
    }
}
